﻿
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace JR.Trc.MessageCenter.Lib
{
    public class RabbitMQHelper
    {
        private static ConnectionFactory _connectionFactory = null;
        static RabbitMQHelper()
        {
            _connectionFactory = new ConnectionFactory();
            _connectionFactory.HostName = ConfigurationManager.AppSettings["HostName"].ToString();
            _connectionFactory.UserName = ConfigurationManager.AppSettings["UserName"].ToString();
            _connectionFactory.Password = ConfigurationManager.AppSettings["Password"].ToString();
            _connectionFactory.VirtualHost = ConfigurationManager.AppSettings["VirtualHost"].ToString();
            _connectionFactory.AutomaticRecoveryEnabled = true;
        }
        #region 单消息入对出队
        /// <summary>
        /// 单消息入队
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="message">消息</param>
        public static void Enqueue(string exchangeName, string exchangeType, string routingKey,  dynamic message)
        {
            try
            {
                if (message != null)
                {
                    using (IConnection connection = _connectionFactory.CreateConnection())
                    {
                        using (IModel channel = connection.CreateModel())
                        {
                            //channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
                            channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
                            string messageString = message.GetType() == typeof(string) ? message : JsonConvert.SerializeObject(message);
                            byte[] body = Encoding.UTF8.GetBytes(messageString);
                            var properties = channel.CreateBasicProperties();
                            properties.Persistent = true; //使消息持久化
                            channel.BasicPublish(exchangeName, routingKey, properties, body);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }

        /// <summary>
        /// 单消息出对
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="exchangeType"></param>
        /// <param name="routingKey"></param>
        /// <param name="queueName"></param>
        /// <param name="func"></param>
        /// <param name="action"></param>
        public static void Consume(string exchangeName, string exchangeType, string routingKey, string queueName, Func<string, bool> func, Action action = null)
        {
            try
            {
                bool isConsumeSuccess = false;// 是否消费成功
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//告诉broker同一时间只处理一个消息
                        //channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
                        channel.QueueDeclare(queueName, true, false, false, null);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                        consumer.Received += (sender, eventArgs) =>
                        {
                            byte[] body = eventArgs.Body;
                            if (body != null && body.Length > 0)
                            {
                                string message = Encoding.UTF8.GetString(body);
                                if (!string.IsNullOrWhiteSpace(message))
                                {
                                    isConsumeSuccess = func(message);
                                    if (isConsumeSuccess)
                                    {
                                        channel.BasicAck(eventArgs.DeliveryTag, false);
                                    }
                                }
                            }
                        };
                        //（第二个参数false的含义打开消息应答机制）
                        channel.BasicConsume(queueName, false, consumer);
                        if (action != null)
                        {
                            action();
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }
        #endregion


        #region 消息批量入队，出对
        /// <summary>
        /// 消息批量入队
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="list">消息集合</param>
        public static void Enqueue<TItem>(string exchangeName, string exchangeType, string routingKey, List<TItem> list)
        {
            try
            {
                if (list != null && list.Count > 0)
                {
                    using (IConnection connection = _connectionFactory.CreateConnection())
                    {
                        using (IModel channel = connection.CreateModel())
                        {
                            foreach (TItem item in list)
                            {
                                if (item != null)
                                {
                                    channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
                                    string messageString = JsonConvert.SerializeObject(item);
                                    byte[] body = Encoding.UTF8.GetBytes(messageString);
                                    var properties = channel.CreateBasicProperties();//使消息持久化
                                    properties.Persistent = true;
                                    channel.BasicPublish(exchangeName, routingKey, properties, body);
                                }
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }

        /// <summary>
        /// 消息出对队列
        /// </summary>
        /// <typeparam name="TItem">消息对象</typeparam>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="func">消费消息的具体操作</param>
        /// <param name="tryTimes">消费失败后，继续尝试消费的次数</param>
        public static void Consume<TItem>(string exchangeName, string exchangeType, string routingKey, string queueName, Func<TItem, bool> func, Action action = null) where TItem : class
        {
            try
            {
                bool isConsumeSuccess = false;// 是否消费成功
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
                        channel.QueueDeclare(queueName, true, false, false, null);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                        consumer.Received += (sender, eventArgs) =>
                        {
                            byte[] body = eventArgs.Body;
                            if (body != null && body.Length > 0)
                            {
                                string message = Encoding.UTF8.GetString(body);
                                if (!string.IsNullOrWhiteSpace(message))
                                {
                                    TItem queueMessage = JsonConvert.DeserializeObject<TItem>(message);
                                    if (queueMessage != null)
                                    {
                                        isConsumeSuccess = func(queueMessage);
                                        if (isConsumeSuccess)
                                        {
                                            channel.BasicAck(eventArgs.DeliveryTag, false);
                                        }
                                    }
                                }
                            }
                        };
                        channel.BasicConsume(queueName, false, consumer);
                        if (action != null)
                        {
                            action();
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }

        #endregion
    }
}
